package com.demo.config;

import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.HashMap;
import java.util.Map;

/**
 * ttl队列 + 死信队列 案例
 *
 * @author jie.luo
 * @since 2021/3/1
 */
@Configuration
public class RabbitMqTtlConfig {

    /**
     * 定义 下单交换器
     */
    @Bean
    public Exchange orderExchange() {
        return new DirectExchange("order.exchange", true, false, null);
    }

    /**
     * 定义 下单消息队列
     */
    @Bean
    public Queue orderQueue() {
        return new Queue("order.queue", true, false, false, null);
    }

    /**
     * 绑定 下单交换器 、下单消息队列
     */
    @Bean
    public Binding orderBinding() {
        return BindingBuilder.bind(orderQueue()).to(orderExchange()).with("order.key").noargs();
    }

    /**
     * 定义 下单延时交换器
     */
    @Bean
    public Exchange orderExchangeTtl() {
        return new DirectExchange("order.exchange.ttl", true, false, null);
    }

    /**
     * 定义 下单延时消息队列
     * 设置 该队列 ttl 为 15s
     */
    @Bean
    public Queue orderQueueTtl() {
        Map<String, Object> arguments = new HashMap<>();
        // 设置队列的ttl
        arguments.put("x-message-ttl", 15 * 1000);
        // 设置该队列关联的死信交换器
        arguments.put("x-dead-letter-exchange", "order.exchange.dle");
        // 设置该队列关联死信交换器的 routing-key
        arguments.put("x-dead-letter-routing-key", "order.key.dle");
        return new Queue("order.queue.ttl", true, false, false, arguments);
    }

    /**
     * 绑定 下单延时交换器 、下单延时消息队列
     */
    @Bean
    public Binding orderBindingTtl() {
        return BindingBuilder.bind(orderQueueTtl()).to(orderExchangeTtl()).with("order.key.ttl").noargs();
    }

    /**
     * 定义 下单消息关联的 死信交换器
     */
    @Bean
    public Exchange orderDeadExchange() {
        return new DirectExchange("order.exchange.dle", true, false, null);
    }

    /**
     * 定义 下单消息队列 关联 的死信队列
     */
    @Bean
    public Queue orderDeadQueue() {
        return new Queue("order.queue.dle", true, false, false, null);
    }

    /**
     * 绑定 下单死信交换器 、下单死信消息队列
     */
    @Bean
    public Binding orderDeadBinding() {
        return BindingBuilder.bind(orderDeadQueue()).to(orderDeadExchange()).with("order.key.dle").noargs();
    }

    @Bean
    public AmqpAdmin amqpAdmin(ConnectionFactory connectionFactory) {
        return new RabbitAdmin(connectionFactory);
    }

    @Bean
    public RabbitTemplate jacksonRabbitTemplate(ConnectionFactory connectionFactory, ObjectMapper objectMapper) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter(objectMapper));
        return rabbitTemplate;
    }

    @Bean
    public MessageConverter messageConverter(ObjectMapper objectMapper) {
        return new Jackson2JsonMessageConverter(objectMapper);
    }

    @Bean
    public RabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory, ObjectMapper objectMapper) {
        // SimpleRabbitListenerContainerFactory发现消息中有content_type有text就会默认将其
        // 转换为String类型的，没有content_type都按byte[]类型
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setMessageConverter(messageConverter(objectMapper));
        // 设置消息确认为手动
        factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        // 设置并发线程数
        factory.setConcurrentConsumers(10);
        // 设置最大并发线程数
        factory.setMaxConcurrentConsumers(20);
        return factory;
    }
}
